package com.cssl.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import com.cssl.app.function.CustomerDeserialization;
import com.cssl.app.function.DimSinkFunction;
import com.cssl.app.function.TableProcessFunction;
import com.cssl.bean.TableProcess;
import com.cssl.utils.CommonUtils;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.flink.util.OutputTag;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.util.Properties;

/**
 * @Author: chen
 * @Date: 2021/11/11 23:23
 * @Desc: 获取 kafka ods topic 数据 , 将 dim 数据持久化到 phoenix , 将 dwd 数据发送到 kafka dwd topic
 */
public class BaseDBApp {
    public static void main(String[] args) throws Exception {
        //1. 获取执行环境
        Properties properties = CommonUtils.getProperties();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2. 消费 kafka ods_base_db 主题数据创建流
        String topic = properties.getProperty("kafka.topic.db.ods");
        DataStreamSource<String> kafkaSourceDS = env.addSource(CommonUtils.getKafkaConsumer(topic));

        //3. 将每行数据转换为 json 对象并过滤(delete)
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaSourceDS.map(JSON::parseObject)
                .filter(jsonObj -> {
                    String type = jsonObj.getString("type");
                    return !"delete".equals(type);
                });

        //4. 使用 flink cdc 消费配置表并处理成广播流
        DebeziumSourceFunction<String> sourceFunction = MySQLSource.<String>builder()
                .hostname("hadoop102")
                .port(3306)
                .username("root")
                .password("root")
                .databaseList("gmall_realtime")
                .tableList("gmall_realtime.table_process")
                .startupOptions(StartupOptions.initial())
                .deserializer(new CustomerDeserialization())
                .build();
        DataStreamSource<String> tableProcessStrDS = env.addSource(sourceFunction);

        MapStateDescriptor<String, TableProcess> mapStateDescriptor = new MapStateDescriptor<>("map-state", String.class, TableProcess.class);
        BroadcastStream<String> broadcastStream = tableProcessStrDS.broadcast(mapStateDescriptor);

        //5. 连接主流和广播流
        BroadcastConnectedStream<JSONObject, String> connectedStream = jsonObjDS.connect(broadcastStream);

        //6. 分流 处理数据 广播流数据,主流数据(根据广播流数据进行处理)
        OutputTag<JSONObject> hbaseTag = new OutputTag<JSONObject>("hbase-tag") {
        };
        SingleOutputStreamOperator<JSONObject> kafkaSinkDS = connectedStream.process(new TableProcessFunction(hbaseTag, mapStateDescriptor));

        //7. 提取 kafka 流数据和 hbase 流数据
        DataStream<JSONObject> hbaseSinkDS = kafkaSinkDS.getSideOutput(hbaseTag);

        //8. 将 kafka 数据写入 kafka ,将 hbase 数据写入 phoenix 表
        hbaseSinkDS.addSink(new DimSinkFunction());

        kafkaSinkDS.addSink(CommonUtils.getKafkaProducer(new KafkaSerializationSchema<JSONObject>() {
            @Override
            public ProducerRecord<byte[], byte[]> serialize(JSONObject jsonObject, @Nullable Long aLong) {
                return new ProducerRecord<byte[], byte[]>(
                        jsonObject.getString("sinkTable"),
                        jsonObject.getString("after").getBytes());
            }
        }));

        //9. 启动任务
        env.execute("BaseDBApp");
    }
}
